package org.fhm.zdte.common.pool;

import org.fhm.substrate.annotation.BeanInitial;
import org.fhm.substrate.annotation.Component;
import org.fhm.substrate.exception.NormalException;
import org.fhm.substrate.service.LoggerHandler;
import org.fhm.substrate.standard.ILogger;
import org.fhm.substrate.util.IOCExceptionUtil;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * The customized thread pool factory.
 * <p>The thread pool factory of Zero-Distance.</p>
 * <p>The thread pool factory used to create
 * {@link DistributorThreadPool}, {@link TransformMessageThreadPool}
 * and used to manage these thread pool.</p>
 *
 * @author 谭波
 * @since 2023/11/4
 */
@Component
public class ThreadPoolFactory {

    public static final byte DISTRIBUTOR_THREAD_POOL = 0b00000001;

    public static final byte TRANSFORM_THREAD_POOL = 0b00000010;

    private final ILogger logger = LoggerHandler.getLogger(ThreadPoolFactory.class);

    private final Map<Byte, AbstractThreadPoolExecutor> tlc = new HashMap<>(2, 1);

    private ScheduledThreadPool scheduledThreadPool;

    /**
     * Execute the schedule task of the specified thread pool
     *
     * @param task  run
     * @param delay execute time delay
     * @param unit  seconds or minutes
     * @return ScheduledFuture
     */
    public ScheduledFuture<?> submitPeriodicTask(
            Runnable task,
            long delay, TimeUnit unit
    ) throws NormalException {
        return scheduledThreadPool.scheduleWithFixedDelay(task, 0, delay, unit);
    }

    /**
     * Submit a task to specified thread pool.
     *
     * @param threadPoolType Thread pool type.
     * @param task           Task.
     */
    public void submitTask(byte threadPoolType, Runnable task) {
        AbstractThreadPoolExecutor executor = tlc.get(threadPoolType);
        if (Objects.isNull(executor)) {
            throw IOCExceptionUtil.generateNormalException(threadPoolType + " is illegal flag");
        }
        executor.submit(task);
    }

    /**
     * Close all the thread pool of Zero-Distance.
     */
    public void closeThreadPool() {
        tlc.values().stream()
                .filter(pool -> !pool.isShutdown())
                .forEach(this::closeThreadPool);
        closeThreadPool(scheduledThreadPool);
    }

    /**
     * Close specified {@link ExecutorService}
     *
     * @param pool thread pool object
     */
    public void closeThreadPool(ExecutorService pool) {
        pool.shutdown();
        try {
            for (int i = 0; i < 3; i++) {
                if (pool.awaitTermination(1, TimeUnit.SECONDS)) {
                    return;
                }
            }
            if (!pool.isTerminated()) {
                pool.shutdownNow();
            }
        } catch (InterruptedException e) {
            logger.warn("pool close fail {} {}", e.getMessage(), e.getClass());
        }
    }

    /**
     * Initialize the thread pool of Zero-Distance.
     */
    @BeanInitial
    private void beanInitial() {
        Runtime runtime = Runtime.getRuntime();
        int processors = runtime.availableProcessors();
        int keepAliveTime = 300;
        int size = processors / 8;
        int baseSize = size == 0 ? 1 : size;
        long memorySizeBKB = runtime.maxMemory() / 1024 / 5;
        int coreSize = baseSize * 2;
        int maxSize = coreSize * 2;
        scheduledThreadPool = new ScheduledThreadPool(
                coreSize, "schedule-",
                maxSize, 300, new ThreadPoolExecutor.AbortPolicy()
        );
        tlc.put(
                DISTRIBUTOR_THREAD_POOL,
                new DistributorThreadPool(
                        coreSize,
                        maxSize,
                        keepAliveTime,
                        new LinkedBlockingDeque<>((int) (memorySizeBKB))
                )
        );
        int transformCoreSize = processors - 1 - coreSize * 2;
        transformCoreSize = transformCoreSize <= 0 ? processors : transformCoreSize;
        tlc.put(
                TRANSFORM_THREAD_POOL,
                new TransformMessageThreadPool(
                        transformCoreSize,
                        transformCoreSize * 2,
                        keepAliveTime,
                        new LinkedBlockingDeque<>((int) (memorySizeBKB * 2))
                )
        );
    }
}
